package beautiful.butterfly.drds.data_exchange.core.statistics.plugin.task;


import beautiful.butterfly.drds.data_exchange.core.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.core.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.core.util.FrameworkErrorCode;
import common.constant.PluginType;
import common.element.Record;
import common.exception.DataExchangeException;
import common.plugin.TaskPluginCollector;
import common.util.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public abstract class AbstractTaskPluginCollector extends TaskPluginCollector
{
    private static final Logger LOG = LoggerFactory
            .getLogger(AbstractTaskPluginCollector.class);

    private Communication communication;

    private Configuration configuration;

    private PluginType pluginType;

    public AbstractTaskPluginCollector(Configuration conf, Communication communication,
                                       PluginType type)
    {
        this.configuration = conf;
        this.communication = communication;
        this.pluginType = type;
    }

    public Communication getCommunication()
    {
        return communication;
    }

    public Configuration getConfiguration()
    {
        return configuration;
    }

    public PluginType getPluginType()
    {
        return pluginType;
    }

    @Override
    final public void collectMessage(String key, String value)
    {
        this.communication.addMessage(key, value);
    }


    public void collectDirtyRecord(Record dirtyRecord, Throwable t,
                                   String errorMessage)
    {

        if (null == dirtyRecord)
        {
            LOG.warn("脏数据record=null.");
            return;
        }

        if (this.pluginType.equals(PluginType.reader))
        {
            this.communication.increaseCounter(
                    CommunicationTool.read_failed_records, 1);
            this.communication.increaseCounter(
                    CommunicationTool.read_failed_bytes, dirtyRecord.getByteSize());
        } else if (this.pluginType.equals(PluginType.writer))
        {
            this.communication.increaseCounter(
                    CommunicationTool.write_failed_records, 1);
            this.communication.increaseCounter(
                    CommunicationTool.write_failed_bytes, dirtyRecord.getByteSize());
        } else
        {
            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR,
                    String.format("不知道的插件类型[%s].", this.pluginType));
        }
    }
}
